package org.ns.longhttp.handler.asynchandler;

import org.ns.longhttp.BeanUtil;
import org.ns.longhttp.ThreanName;
import org.ns.longhttp.config.ServerConfig;
import org.ns.longhttp.selector.datamapper.DataMappingEntry;
import org.ns.longhttp.selector.datamapper.DataMappingQueue;
import org.ns.longhttp.selector.datamapper.DataResponseEntry;
import org.ns.longhttp.selector.requestmapper.RequestMappingQueue;
import org.ns.longhttp.wrapper.GetRequestWrapper;
import org.ns.longhttp.wrapper.GetResponseWrapper;

import java.util.*;
import java.util.concurrent.*;

/**
 * 异步数据应答处理
 */
public class AsyncResponseHandler {


    private DataMappingQueue dataQueue;
    private RequestMappingQueue requestQueue;
    // 数据应答事件队列
    private UrlEventQueue urlQueue;
    // private ConcurrentHashMap<String, Boolean> urlQueueDuplicate = new ConcurrentHashMap<>();
    // 监听线程
    private ExecutorService listenerThread = Executors.newFixedThreadPool(1, new ThreanName("AsyncResponseListener", true));
    // 监听事件-提交事件的工作
    private ExecutorService listenerThreadSubmit;

    public AsyncResponseHandler(DataMappingQueue dataQueue, RequestMappingQueue requestQueue) {
        this.dataQueue = dataQueue;
        this.requestQueue = requestQueue;
    }
    public void add(String url) {
        //if (urlQueueDuplicate.putIfAbsent(url, true) == null) {
        urlQueue.add(url);
        //}
    }
    public void setDataQueue(DataMappingQueue dataQueue) {
        this.dataQueue = dataQueue;
    }

    public void setRequestQueue(RequestMappingQueue requestQueue) {
        this.requestQueue = requestQueue;
    }

    public void setUrlQueue(UrlEventQueue urlQueue) {
        this.urlQueue = urlQueue;
    }

    private void init() {
        listenerThread.execute(this::listenerThreadRun);
    }

    private void listenerThreadRun(){
        while (true) {
            try {
                String url = urlQueue.pull(5);
                if (url != null) {
                    Set<String> urls = new HashSet<>();
                    urls.add(url);

                    int size = urlQueue.size();
                    for (int i = 0; i < size; i++) {
                        url = urlQueue.pop();
                        urls.add(url);
                    }
                    List<Future<?>> futures = new ArrayList<>(urls.size());
                    for (String tmp : urls) {
                        Future<?> future = listenerThreadSubmit.submit(()->{
                            submitResponse(tmp);
                        });
                    }
                    for (Future<?> future : futures) {
                        try {
                            future.get();
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    }
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }


    private void submitResponse(String url) {
//        urlQueueDuplicate.remove(url);
        urlQueue.consume(url); // 消费这个事件
        DataMappingEntry dataMappingEntry = dataQueue.getDataMappingEntry(url);
        List<GetRequestWrapper> requestWrappers = requestQueue.getRequestWrappers(url);

        for (GetRequestWrapper requestWrapper : requestWrappers) {
            long id = requestWrapper.getDataid();
            // System.out.println("readL" + id);
            List<DataResponseEntry> dataResponseEntrys = dataMappingEntry.getDataResponseEntrys(id);
            if (dataResponseEntrys.size() > 0) {
                GetResponseWrapper responseWrapper = new GetResponseWrapper(dataResponseEntrys);
                requestWrapper.write(responseWrapper);
            }
        }
    }
    // 初始化配置
    public void config(ServerConfig serverConfig) {
        // 异步应答提交使用的线程数
        int asyncResponseThreadSize = serverConfig.getAsyncResponseThreadSize();
        listenerThreadSubmit = Executors.newFixedThreadPool(asyncResponseThreadSize, new ThreanName("AsyncResponse-submit", true));
        // 事件队列实现类
        String urlEventQueueClassname = serverConfig.getUrlEventQueueClassname();
        this.urlQueue = (UrlEventQueue)BeanUtil.newInstance(urlEventQueueClassname);
        // 初始化异步应答处理器
        this.init();
    }
}
